package com.caption.mq.core;

import com.caption.mq.entity.ModelResult;
import com.caption.mq.entity.MqMessageDTO;
import com.caption.mq.entity.MqSubscribeDO;
import com.caption.mq.listener.AbstractMqListener;
import lombok.Getter;
import lombok.Setter;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMqConsumer extends AbstractMqConsumer {

    private DefaultMQPushConsumer mqPushConsumer;

    private final String nameServeAddress;
    /**
     * 异常情况集群消费重试策略，值等于0每隔5秒钟重试
     * 值等于1重试时间按 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 的
     * 间隔衰减重试。
     */
    @Setter
    @Getter
    private int failRetryStrategy = 0;
    /**
     * 集群消费重试次数,默认3次
     */
    @Setter
    @Getter
    private int retryCount = 3;

    @Setter
    @Getter
    private Boolean started;
    /**
     * 消息延时等级
     */
    private int delayLevelWhenNextConsume = 0;
    /**
     * 消息延迟发送时间
     */
    private int suspendCurrentQueueTimeMillis = 1000;

    public RocketMqConsumer(MqSubscribeDO subscribe, String nameServeAddress) {
        super(subscribe);
        this.nameServeAddress = nameServeAddress;
        this.retryCount = subscribe.getRetryCount();

        //消息重新策略 0-每隔5秒重试， 1-重试时间按 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 增加
        if (subscribe.getFailRetryStrategy() == 0) {
            this.delayLevelWhenNextConsume = 2;
        } else {
            this.delayLevelWhenNextConsume = 0;
        }
        initConsumer(subscribe, nameServeAddress);
    }

    /**
     * 初始化 DefaultMQPushConsumer
     *
     * @param subscribe
     * @param nameServeAddress
     */
    public void initConsumer(MqSubscribeDO subscribe, String nameServeAddress) {
        mqPushConsumer = new DefaultMQPushConsumer(subscribe.getGroup());
        mqPushConsumer.setNamesrvAddr(nameServeAddress);
        mqPushConsumer.setConsumeThreadMax(subscribe.getConsumeThreadMax());
        mqPushConsumer.setConsumeThreadMin(subscribe.getConsumeThreadMin());
        mqPushConsumer.setPullBatchSize(subscribe.getPullBatchSize());
        mqPushConsumer.setConsumeMessageBatchMaxSize(subscribe.getConsumeMessageBatchMaxSize());
        //最大重试次数，超过最大重试次数，消息将被转 移到私信队列
        mqPushConsumer.setMaxReconsumeTimes(subscribe.getRetryCount());
        switch (subscribe.getConsumeFromWhere()) {
            case 0:
                mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
                break;
            case 1:
                mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
                break;
            case 2:
                mqPushConsumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP);
                if (subscribe.getConsumeFromTime() != null) {
                    mqPushConsumer.setConsumeTimestamp(UtilAll.timeMillisToHumanString3(subscribe.getConsumeFromTime().getTime()));
                }
                break;
            default:
                break;
        }
    }

    @Override
    public void subscribe(String topic, String tag) {
        try {
            mqPushConsumer.subscribe(topic, tag);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void registerMessageListener(AbstractMqListener listener) {
        if (this.subscribe.getConsumeMode() == 0) {
            mqPushConsumer.registerMessageListener(new DefaultMessageListenerConcurrently(listener));
        } else {
            mqPushConsumer.registerMessageListener(new DefaultMessageListenerOrderly(listener));
        }
    }

    /**
     * 普通消息
     */
    public class DefaultMessageListenerConcurrently implements MessageListenerConcurrently {
        private AbstractMqListener mqListener;

        public DefaultMessageListenerConcurrently(AbstractMqListener mqListener) {
            super();
            this.mqListener = mqListener;
        }

        @Override
        public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
            MqMessageDTO messageDTO = null;
            for (MessageExt messageExt : msgs) {
                try {
                    messageDTO = getMessageDTO(messageExt);
                    ModelResult result = mqListener.handleMessage(messageDTO);
                    if (result.isSuccess()) {
                        mqListener.onMessageHandleSuccess(messageExt, messageDTO, result);
                    } else {
                        //消息处理失败请继续重试
                        dealFailed(context, messageExt, messageDTO, result);
                    }
                } catch (Exception e) {
                    //出现异常请重试a
                    dealException(context, messageExt, messageDTO, e);
                } finally {

                }
            }
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }

        private ConsumeConcurrentlyStatus dealFailed(ConsumeConcurrentlyContext context, MessageExt messageExt, MqMessageDTO messageDTO, ModelResult modelResult) {
            int reconsumeTimes = messageExt.getReconsumeTimes();
            if (reconsumeTimes < retryCount) {
                context.setDelayLevelWhenNextConsume(RocketMqConsumer.this.delayLevelWhenNextConsume);
            } else {
                mqListener.onMessageHandleFailed(messageExt, messageDTO, modelResult);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }


        private ConsumeConcurrentlyStatus dealException(ConsumeConcurrentlyContext context, MessageExt messageExt, MqMessageDTO messageDTO, Exception e) {
            int reconsumeTimes = messageExt.getReconsumeTimes();
            if (reconsumeTimes < retryCount) {
                context.setDelayLevelWhenNextConsume(RocketMqConsumer.this.delayLevelWhenNextConsume);
            } else {
                mqListener.onMessageHandleException(messageExt, messageDTO, e);
            }
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
    }

    /**
     * 顺序消息，会一直等待前面的消息执行成功才会继续执行
     */

    public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
        private AbstractMqListener mqListener;

        public DefaultMessageListenerOrderly(AbstractMqListener mqListener) {
            super();
            this.mqListener = mqListener;
        }


        @Override
        public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
            for (MessageExt messageExt : msgs) {
                MqMessageDTO messageDTO = null;
                try {
                    messageDTO = getMessageDTO(messageExt);
                    ModelResult modelResult = mqListener.handleMessage(messageDTO);
                    if (modelResult.isSuccess()) {
                        mqListener.onMessageHandleSuccess(messageExt, messageDTO, modelResult);
                    } else {
                        context.setSuspendCurrentQueueTimeMillis(10);
                        mqListener.onMessageHandleFailed(messageExt, messageDTO, modelResult);

                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                    }
                } catch (Exception e) {
                    context.setSuspendCurrentQueueTimeMillis(10);
                    mqListener.onMessageHandleException(messageExt, messageDTO, e);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }
            return ConsumeOrderlyStatus.SUCCESS;
        }
    }


    /**
     * 转换请求的实体
     *
     * @param messageExt
     * @return
     */
    public MqMessageDTO getMessageDTO(MessageExt messageExt) {
        return null;
    }


    /**
     * 启动监听
     */
    @Override
    public void start() {
        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        this.setStarted(true);
    }

    /**
     * 关闭监听
     */
    @Override
    public void shutdown() {
        this.setStarted(false);
        mqPushConsumer.unsubscribe(this.subscribe.getTopic());
        mqPushConsumer.shutdown();

    }
}
